﻿using Basic.EventBus.Attributes;
using Basic.EventBus.Eventbus;
using Confluent.Kafka;
using Microsoft.Extensions.Logging;

namespace Basic.EventBus.EventBusKafka
{
    public class KafkaProducer : IEventPublisher
    {
        private ILogger<KafkaProducer> Logger;
        private KafkaBusBrokerConfig BrokerConfig;
        private IProducer<string, byte[]> Producer;

        public KafkaProducer(ILogger<KafkaProducer> logger, KafkaBusBrokerConfig options)
        {
            Logger = logger;
            BrokerConfig = options;

            var conf = new ProducerConfig { BootstrapServers = BrokerConfig.Servers ?? ($"{BrokerConfig.Host}:{BrokerConfig.Port}") };
            var builder = new ProducerBuilder<string, byte[]>(conf);
            Producer = builder.Build();
        }

        public async Task<bool> PublishAsync(byte[] data, EventTopicAttribute attribute)
        {
            var message = new Message<string, byte[]> { Key = attribute.Key, Value = data.ToArray() };
            var resp = await Producer.ProduceAsync(attribute.Topic, message);
            return resp.Status > 0;
        }

        public async Task<bool> PublishAsync(IEventData eventData)
        {
            var attribute = EventTopicAttribute.GetCurrentAttribute(eventData.GetType());
            if (attribute == null)
                return false;
            return await PublishAsync(eventData.Serialize(), attribute);
        }

        public async Task<int> PublishAsync(List<IEventData> eventDatas)
        {
            var attribute = EventTopicAttribute.GetCurrentAttribute(eventDatas[0].GetType());
            if (attribute == null) return 0;

            foreach (var eventData in eventDatas)
                await PublishAsync(eventData.Serialize(), attribute);

            return eventDatas.Count;
        }

        #region Dispose
        private bool disposedValue;

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    Producer.Dispose();
                }
                // TODO: 将大型字段设置为 null
                Producer = null;
                disposedValue = true;
            }
        }

        public void Dispose()
        {
            // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
        #endregion

    }
}